package com.tc.vms.consumer;

import com.tc.vms.SLA;
import com.tc.vms.VMSEventArgs;
import com.tc.vms.connection.VmsConnectionManager;
import com.tc.vms.exception.VmsClientErrorCode;
import com.tc.vms.exception.VmsClientException;
import com.tc.vms.iface.*;
import com.tc.vms.utils.Utils;
import com.yammer.metrics.core.Timer;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * Created by yonghua.zhang on 2016/1/7.
 */
public class VmsConsumerProxy extends VmsConnectionWrap implements ISubscriber {

    private volatile ISubscriber subscriber;
    private volatile IVMSCallbackProxy ivmsCallbackProxy;
    private final VmsConsumerMetrics vmsConsumerMetrics;
//    private final VmsHermesManager.VmsHermesConsumerStat vmsHermesConsumerStat;

    public VmsConsumerProxy(VmsConnectionManager vmsConnectionManager, VmsMetaData vmsMetaData, SLA serviceSLA, VmsQos vmsQos, Properties props) {
        super(vmsConnectionManager, vmsMetaData, serviceSLA, vmsQos, props);
        vmsConsumerMetrics = getConsumerMetrics();
//        vmsHermesConsumerStat = new VmsHermesManager.VmsHermesConsumerStat(vmsMetaData.getOrigClientId(),
//                Utils.getHostIp(), vmsMetaData.getZnodeName(), vmsConsumerMetrics);
    }

    @Override
    protected double getRealTpsInOneMinute() {
        return Math.max(vmsConsumerMetrics.consumeSuccessRate().oneMinuteRate(),
                vmsConsumerMetrics.consumeFailureRate().oneMinuteRate());
    }

    public VmsConsumerProxy(VmsConnectionManager vmsConnectionManager, VmsMetaData vmsMetaData, Properties props) {
        this(vmsConnectionManager, vmsMetaData, new SLA(), new SubQoS(), props);
    }

    public class IVMSCallbackProxy implements IVMSCallback {
        private final IVMSCallback callback;

        public IVMSCallbackProxy(IVMSCallback callback) {
            this.callback = callback;
        }

        @Override
        public ConsumeStatus consumeMessage(VmsConsumerContext context, VMSEventArgs args) {
            Timer totalTimer = vmsConsumerMetrics.totalCostOfMessage();
            Timer UserTimer = vmsConsumerMetrics.userCostOfMessage();
            long currentTimestamp = System.currentTimeMillis();
            ConsumeStatus cs = null;
            try {
                LOGGER.debug("VmsConsumerProxy begin to consume the message. ackIdx=[{}], messageid=[{}].", context.getAckIdx(), args.getMessage().getMessageId());
                cs = callback.consumeMessage(context, args);
            } catch (Throwable e) {
                LOGGER.warn(String.format("consumeMessage failure. ackIdx=[%d], messageid=[%s]", context.getAckIdx(), args.getMessage().getMessageId()), e);
            }
            cs = (cs == null) ? ConsumeStatus.CONSUME_DISCARD : cs;
            UserTimer.update(System.currentTimeMillis() - currentTimestamp, TimeUnit.MILLISECONDS);
            totalTimer.update(System.currentTimeMillis() - args.getMessage().getBornTime(), TimeUnit.MILLISECONDS);

            if (!cs.equals(ConsumeStatus.CONSUME_SUCCESS)) {
                vmsConsumerMetrics.consumeFailureRate().mark();
                if (args.getMessage().getReconsumeTimes() > 0) {
                    vmsConsumerMetrics.reconsumeFailureRate().mark();
                }
            } else {
                vmsConsumerMetrics.consumeSuccessRate().mark();
                if (args.getMessage().getReconsumeTimes() > 0) {
                    vmsConsumerMetrics.reconsumeSuccessRate().mark();
                }
            }
            LOGGER.debug("VmsConsumerProxy have consumed the message of id=[{}], status=[{}].", args.getMessage().getMessageId(), cs);
            return cs;
        }
    }

    @Override
    public synchronized void registerListener(IVMSCallback callback) {
        ivmsCallbackProxy = new IVMSCallbackProxy(callback);
        subscriber.registerListener(ivmsCallbackProxy);
    }

    @Override
    public void suspend() {
        this.isSuspend = true;
    }

    @Override
    public void resume() {
        this.isSuspend = false;
    }

    @Override
    public void setConsumingThreadNum(int threadNum) {
        if (subscriber != null) {
            subscriber.setConsumingThreadNum(threadNum);
        }
    }

    /**
     * for outer user call, inner invoke need using sla() or member directly
     *
     * @return
     */
    public final SLA slaOption() {
        try {
            chkResetOption();
            return super.sla();
        } catch (VmsClientException e) {
            LOGGER.warn("Vmsclient is running,couldn't modify service option.", e);
        }

        return super.sla().cloneSLA(); // return new option to avoid NPE
    }

    public final SubQoS qosOption() {
        try {
            chkResetOption();
            return super.qos();
        } catch (VmsClientException e) {
            LOGGER.warn("Vmsclient is running,couldn't modify service option.", e);
        }
        return ((SubQoS) super.qos()).cloneSubQoS(); // return new option to avoid NPE
    }

    @Override
    public synchronized boolean init(IContext context) throws VmsClientException {
        if (subscriber == null) {
            QueueMetaData queueMetaData = getVmsMetaData();
            switch (queueMetaData.getBrokerType()) {
                case KAFKA:
                    LOGGER.info("VmsConsumerProxy initialize Kafka consumer instance. ");
                    if (queueMetaData.isDlqQueueMetaData()) {
                        LOGGER.error("Kafka don't support DLQ_queue subscriber [{}].", queueMetaData);
                        throw new VmsClientException("Kafka don't support DLQ_queue subscriber.");
                    } else {
                        subscriber = new VmsConsumerKafkaImpl(this);
                    }
                    break;
                case RABBITMQ:
                    if (queueMetaData.isDlqQueueMetaData()) {
                        LOGGER.info("VmsConsumerProxy initialize Rabbitmq QLQ consumer instance [{}].", queueMetaData);
                        subscriber = new VmsDLQConsumerRmqImpl(this);
                    } else {
                        LOGGER.info("VmsConsumerProxy initialize Rabbitmq consumer instance. ");
                        subscriber = new VmsConsumerRmqImpl(this);
                    }
                    break;
                default:
                    throw new VmsClientException("VmsConsumerFactory don't support subscriber type [" + queueMetaData.getBrokerType() + "].",
                            VmsClientErrorCode.VMS_SYS_ERR_PRODUCER_TYPE_NOT_SUPPORT);
            }
            subscriber.init(context);
        }
        return true;
    }

    @Override
    public synchronized boolean start() throws VmsClientException {
        if (!isOpen) {
            if (ivmsCallbackProxy == null) {
                throw new VmsClientException("Please set callback before start consuming.");
            }
            subscriber.start();
            registerZkNode();
//            vmsHermesConsumerStat.registerHermes();
            isOpen = true;
            LOGGER.info("VmsConsumerProxy [{}] start successfully. ", vmsMetaData.getClientInstName());
        }
        return isOpen;
    }

    @Override
    public synchronized void shutdown() {
        if (isOpen) {
            isOpen = false;
            try {
                super.shutdown();
                subscriber.shutdown();
                vmsConnectionManager.getVmsMetricsManager().removeVmsMetrics(vmsMetaData);
//                vmsHermesConsumerStat.unRegisterHermes();
            } catch (Exception ex) {
            }
            LOGGER.info("VmsConsumerProxy [{}] shutdown successfully. ", vmsMetaData.getClientInstName());
        }
    }

    @Override
    public synchronized void zkNotifyReconnectServer(VmsMetaData newMetaData, VmsMetaData origMetaData) {
        this.vmsConnectionManager.updateConnectionWrap(this, origMetaData, newMetaData);
        if (subscriber != null) {
            subscriber.shutdown();
            subscriber = null;
        }
        int retryTimes = 0;
        boolean rebootOk = false;
        while (isOpen && retryTimes++ < 3) {
            LOGGER.debug("VmsConsumerProxy zkNotifyReconnectServer restart service for [{}] times.", retryTimes);
            try {
                init(null);
                if (subscriber != null) {
                    subscriber.registerListener(ivmsCallbackProxy);
                    subscriber.start();
                } else {
                    throw new VmsClientException("zkNotifyReconnectServer Init subscriber failed.");
                }
                rebootOk = true;
                break;
            } catch (VmsClientException e) {
                LOGGER.warn("ReConnectServer error, retry again.", e);
                Utils.sleep3s();
                if (subscriber != null) {
                    subscriber.shutdown();
                    subscriber = null;
                }
            }
        }

        if (!rebootOk) {
            LOGGER.error("zkNotifyReconnectServer service couldn't reboot. please check the config [{}].", newMetaData);
        }
    }

    private VmsConsumerMetrics getConsumerMetrics() {
        return vmsConnectionManager.getVmsMetricsManager().getVmsMetrics(vmsMetaData);
    }
}
